package com.hyt.it.ogt.ks.util;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import java.util.stream.Collectors;

import org.springframework.util.CollectionUtils;

import com.google.common.base.Stopwatch;

import lombok.extern.slf4j.Slf4j;

/**
 * @author wuyingge
 * @ClassName JobThreadPoolTask.java
 * @Description 用与多线程处理数据，等待线程完成，记录失败的数据
 * @createTime 2021年09月17日 18:25:00
 */
@Slf4j
public class JobThreadPoolTask {

    /**
     * 传入并发的线程数量
     * @param jobs         需要处理的数据
     * @param consumer     处理数据的对象
     * @param coreThreads  同时并发处理的线程数
     * @param <T>          需要处理数据的类型
     * @throws Exception
     */
    public static<T> void run(List<T> jobs, Consumer<T> consumer,int coreThreads,String logMsg) throws Exception{
        if(CollectionUtils.isEmpty(jobs)){
            return;
        }
        run(coreThreads,jobs,consumer,logMsg);
    }

    /**
     *
     * @param jobs         需要处理的数据
     * @param consumer     处理数据的对象
     * @param <T>          需要处理数据的类型
     * @throws Exception
     */
    public static<T> void run(List<T> jobs, Consumer<T> consumer,String logMsg) throws Exception{
        if(CollectionUtils.isEmpty(jobs)){
            return;
        }
        run(Runtime.getRuntime().availableProcessors(),jobs,consumer,logMsg);//使用当前系统活动的cpu数作为并发线程数
    }

    private static<T> void run(int coreThreads,List<T> jobs,Consumer<T> consumer,String logMsg)throws Exception{
        log.info("线程池异步执行,一共需要执行{}个 {}任务", jobs.size(),logMsg);
        // 计时器
        Stopwatch mainThread = Stopwatch.createStarted();
        //线程池 当前批次的数量可控制不用担心队列等待数量过大的问题
        final ExecutorService threadPool = Executors.newFixedThreadPool(coreThreads <= 0 ? jobs.size() : coreThreads, r -> {
            Thread t = new Thread(r);
            t.setName("试卷初始化线程" + t.getId());
            return t;
        });
        //线程计数器
        final CountDownLatch countDownLatch = new CountDownLatch(jobs.size());//总共有多少任务要处理
        //报错的任务信息记录
        List<T> errorJobs = new ArrayList<>();

        for(T t : jobs){
            threadPool.submit(() -> {
                try {
                    log.info("任务:{}, {}开始执行。",t,logMsg);
                    //计时器
                    Stopwatch jobThread = Stopwatch.createStarted();
                    //执行目标任务
                    consumer.accept(t);
                    log.info("任务:{}，{}完成，耗时{}秒。",t,logMsg,jobThread.elapsed(TimeUnit.SECONDS));
                } catch (Exception e){
                    log.error("任务:{},{}异常，原因为:{}",t,logMsg,e);
                    //放入异常容器
                    errorJobs.add(t);
                } finally {
                    countDownLatch.countDown();
                }
            });
        }
        //等待所有的线程完成，记录当前失败的任务
        countDownLatch.await();
        threadPool.shutdown();

        if(!errorJobs.isEmpty()){
            String errMsg = errorJobs.parallelStream().map(T::toString).collect(Collectors.joining(","));
            log.error("以下{}任务：{}",logMsg,errMsg);
        }
        log.info("{}线程池任务全部完成，共耗时【{}】秒",logMsg,mainThread.elapsed(TimeUnit.SECONDS));
    }
}
